package com.alibaba.algo.utils;

import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;

import java.io.*;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;


/**
 * Redis client base on jedis
 *
 * @author xuxueli 2015-7-10 18:34:07
 */
public class JedisUtil {
    private static Logger logger = LoggerFactory.getLogger(JedisUtil.class);

    /**
     * redis address, like "{ip}"、"{ip}:{port}"、"{redis/rediss}://xxl-sso:{password}@{ip}:{port:6379}/{db}"；Multiple "," separated
     */
    private static String address;

    private static String pwd;

    public static void init(String address) {
        JedisUtil.address = address;
        getInstance();
    }

    public static void init(String address, String pwd) {
        JedisUtil.address = address;
        JedisUtil.pwd = pwd;
        getInstance();
    }

    // ------------------------ ShardedJedisPool ------------------------
    /**
     * 方式01: Redis单节点 + Jedis单例 : Redis单节点压力过重, Jedis单例存在并发瓶颈 》》不可用于线上
     * new Jedis("127.0.0.1", 6379).get("cache_key");
     * 方式02: Redis单节点 + JedisPool单节点连接池 》》 Redis单节点压力过重，负载和容灾比较差
     * new JedisPool(new JedisPoolConfig(), "127.0.0.1", 6379, 10000).getResource().get("cache_key");
     * 方式03: Redis分片(通过client端集群,一致性哈希方式实现) + Jedis多节点连接池 》》Redis集群,负载和容灾较好, ShardedJedisPool一致性哈希分片,读写均匀，动态扩充
     * new ShardedJedisPool(new JedisPoolConfig(), new LinkedList<JedisShardInfo>());
     * 方式03: Redis集群；
     * new JedisCluster(jedisClusterNodes);    // TODO
     */

    private static ShardedJedisPool shardedJedisPool;
    private static ReentrantLock INSTANCE_INIT_LOCL = new ReentrantLock(false);

    /**
     * 获取ShardedJedis实例
     *
     * @return
     */
    private static ShardedJedis getInstance() {
        if (shardedJedisPool == null) {
            try {
                if (INSTANCE_INIT_LOCL.tryLock(2, TimeUnit.SECONDS)) {

                    try {

                        if (shardedJedisPool == null) {
                            //JedisPoolConfig
                            JedisPoolConfig config = new JedisPoolConfig();
                            //在指定时刻通过pool能够获取到的最大的连接的jedis个数
                            config.setMaxTotal(1000);
                            //最大能够保持空闲的数量，控制一个pool最多有多少个状态为空闲的jedis实例
                            config.setMaxIdle(600);
                            //在容器中的最小的闲置连接数，仅仅在此值为正数且timeBetweenEvictionRunsMillis值大于0时有效,确保在对象逐出线程工作后确保线程池中有最小的实例数量，如果该值设定大于maxidle的值，此值不会生效，maxidle的值会生效
                            config.setMinIdle(60);
                            //当连接池内的连接耗尽时，getBlockWhenExhausted为true时，连接会阻塞，超过了阻塞的时间（设定的maxWaitMillis，单位毫秒）时会报错, 小于零:阻塞不确定的时间,  默认-1
                            config.setMaxWaitMillis(10 * 1000);
                            //在borrow一个jedis实例时，是否提前进行validate操作；如果为true，则得到的jedis实例均是可用的；默认是false
                            config.setTestOnBorrow(false);
                            //如果为true，表示有一个idle object evitor线程对idle object进行扫描，如果validate失败，此object会被从pool中drop掉；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义；默认是false
                            config.setTestWhileIdle(true);
                            //表示idle object evitor两次扫描之间要sleep的毫秒数，逐出扫描的时间间隔（毫秒），如果为负数，则不运行逐出线程，默认为-1
                            config.setTimeBetweenEvictionRunsMillis(30 * 1000);
                            //调用returnObject方法时，是否进行有效检查
                            config.setTestOnReturn(false);
                            //做空闲资源检测时，每次检测资源的个数
                            config.setNumTestsPerEvictionRun(5);
                            //资源池中资源的最小空闲时间（单位为毫秒），达到此值后空闲资源将被移除。
                            config.setMinEvictableIdleTimeMillis(300 * 1000);

                            // JedisShardInfo List
                            List<JedisShardInfo> jedisShardInfos = new LinkedList<JedisShardInfo>();
                            String[] addressArr = address.split(",");
                            for (int i = 0; i < addressArr.length; i++) {
                                JedisShardInfo jedisShardInfo = new JedisShardInfo(addressArr[i]);
                                if (!StringUtils.isEmpty(pwd)) {
                                    logger.info("Redis Pwd:" + pwd);
                                    jedisShardInfo.setPassword(pwd);
                                }
                                jedisShardInfos.add(jedisShardInfo);
                            }
                            shardedJedisPool = new ShardedJedisPool(config, jedisShardInfos);
                            logger.info(">>>>>>>>>>> xxl-sso, JedisUtil.ShardedJedisPool init success.");
                        }

                    } finally {
                        INSTANCE_INIT_LOCL.unlock();
                    }
                }

            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }

        if (shardedJedisPool == null) {
            throw new NullPointerException(">>>>>>>>>>> xxl-sso, JedisUtil.ShardedJedisPool is null.");
        }
        logger.info("###############shardedJedis Waiters num:{}", shardedJedisPool.getNumWaiters());
        logger.info("###############shardedJedis Idle num:{}", shardedJedisPool.getNumIdle());
        logger.info("###############shardedJedis Active num:{}", shardedJedisPool.getNumActive());
        ShardedJedis shardedJedis = shardedJedisPool.getResource();
        return shardedJedis;
    }

    public static void close() throws IOException {
        if (shardedJedisPool != null) {
            shardedJedisPool.close();
        }
    }


    // ------------------------ serialize and unserialize ------------------------

    /**
     * 将对象-->byte[] (由于jedis中不支持直接存储object所以转换成byte[]存入)
     *
     * @param object
     * @return
     */
    public static byte[] serialize(Object object) {
        ObjectOutputStream oos = null;
        ByteArrayOutputStream baos = null;
        try {
            // 序列化
            baos = new ByteArrayOutputStream();
            oos = new ObjectOutputStream(baos);
            oos.writeObject(object);
            byte[] bytes = baos.toByteArray();
            return bytes;
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            try {
                oos.close();
                baos.close();
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            }
        }
        return null;
    }

    /**
     * 将byte[] -->Object
     *
     * @param bytes
     * @return
     */
    public static Object unserialize(byte[] bytes) {
        ByteArrayInputStream bais = null;
        try {
            // 反序列化
            bais = new ByteArrayInputStream(bytes);
            ObjectInputStream ois = new ObjectInputStream(bais);
            return ois.readObject();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            try {
                bais.close();
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            }
        }
        return null;
    }

    // ------------------------ jedis util ------------------------
    /**
     * 存储简单的字符串或者是Object 因为jedis没有分装直接存储Object的方法，所以在存储对象需斟酌下
     * 存储对象的字段是不是非常多而且是不是每个字段都用到，如果是的话那建议直接存储对象，
     * 否则建议用集合的方式存储，因为redis可以针对集合进行日常的操作很方便而且还可以节省空间
     */

    /**
     * Set String
     *
     * @param key
     * @param value
     * @param seconds 存活时间,单位/秒
     * @return
     */
    public static String setStringValue(String key, String value, int seconds) {
        String result = null;
        ShardedJedis client = getInstance();
        try {
            result = client.setex(key, seconds, value);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return result;
    }

    /**
     * Set Object
     *
     * @param key
     * @param obj
     * @param seconds 存活时间,单位/秒
     */
    public static String setObjectValue(String key, Object obj, int seconds) {
        String result = null;
        logger.info("########setObjectValue,key{},obj:{}", key, obj);
        ShardedJedis client = getInstance();
        logger.info("########setObjectValue,client:{}", client);
        try {
            result = client.setex(key.getBytes(), seconds, serialize(obj));
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return result;
    }

    /**
     * Get String
     *
     * @param key
     * @return
     */
    public static String getStringValue(String key) {
        String value = null;
        ShardedJedis client = getInstance();
        try {
            value = client.get(key);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return value;
    }

    /**
     * Get Object
     *
     * @param key
     * @return
     */
    public static Object getObjectValue(String key) {
        Object obj = null;
        ShardedJedis client = getInstance();
        try {
            byte[] bytes = client.get(key.getBytes());
            if (bytes != null && bytes.length > 0) {
                obj = unserialize(bytes);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return obj;
    }

    /**
     * Delete key
     *
     * @param key
     * @return Integer reply, specifically:
     * an integer greater than 0 if one or more keys were removed
     * 0 if none of the specified key existed
     */
    public static Long del(String key) {
        Long result = null;
        ShardedJedis client = getInstance();
        try {

            result = client.del(key);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return result;
    }

    public static Long checkExistsAndDel(String key) {
        //默认不存在
        Long result = -1L;
        ShardedJedis client = getInstance();
        try {
            if (client.exists(key)) {
                result = client.del(key);
            } else {
                logger.info("checkExistsAnddel key:{} not exists", key);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return result;
    }

    /**
     * 批量删除
     *
     * @param keys
     * @return 0：删除失败   -1：不存在    大于0：删除多个
     */
    public static Map<String, Long> batchDel(List<String> keys) {
        Map<String, Long> resultMap = new HashMap<>();
        ShardedJedis client = getInstance();
        try {
            for (String key : keys) {
                boolean isExists = client.exists(key);
                if (!isExists) {
                    logger.info("batchDel key:{} not exists", key);
                    resultMap.put(key, -1L);
                    continue;
                }
                Long result = client.del(key);
                resultMap.put(key, result);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return resultMap;
    }

    private static final int SCAN_COUNT = 20;

    /**
     * 批量删除同一前缀的key
     *
     * @param prefix
     * @return
     */
    public static Set<String> delByPrefix(String prefix) {
        Set<String> allKeys = new HashSet<>();
        ShardedJedis client = getInstance();
        try {
            String scanRet = ScanParams.SCAN_POINTER_START;
            Iterator<Jedis> jedisIterator = client.getAllShards().iterator();
            while (jedisIterator.hasNext()) {
                Jedis jedis = jedisIterator.next();
                do {
                    ScanParams scanParams = new ScanParams();
                    scanParams.match(prefix.concat("*"));
                    scanParams.count(SCAN_COUNT);
                    ScanResult<String> scanResult = jedis.scan(scanRet, scanParams);
                    scanRet = scanResult.getCursor();
                    List<String> keys = scanResult.getResult();
                    if (!ObjectUtils.isEmpty(keys)) {
                        jedis.del(keys.toArray(new String[0]));
                        allKeys.addAll(keys);
                    }
                } while (!scanRet.equals(ScanParams.SCAN_POINTER_START));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return allKeys;
    }

    /**
     * 获取同一前缀的key
     *
     * @param prefix
     * @return
     */
    public static Set<String> getByPrefix(String prefix) {
        Set<String> keys = new HashSet<>();
        ShardedJedis client = getInstance();
        try {
            String scanRet = ScanParams.SCAN_POINTER_START;
            Iterator<Jedis> jedisIterator = client.getAllShards().iterator();
            while (jedisIterator.hasNext()) {
                Jedis jedis = jedisIterator.next();
                do {
                    ScanParams scanParams = new ScanParams();
                    scanParams.match(prefix.concat("*"));
                    scanParams.count(SCAN_COUNT);
                    ScanResult<String> scanResult = jedis.scan(scanRet, scanParams);
                    scanRet = scanResult.getCursor();
                    keys.addAll(scanResult.getResult());
                } while (!scanRet.equals(ScanParams.SCAN_POINTER_START));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return keys;
    }

    /**
     * incrBy i(+i)
     *
     * @param key
     * @param i
     * @return new value after incr
     */
    public static Long incrBy(String key, int i) {
        Long result = null;
        ShardedJedis client = getInstance();
        try {
            result = client.incrBy(key, i);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return result;
    }

    /**
     * exists valid
     *
     * @param key
     * @return Boolean reply, true if the key exists, otherwise false
     */
    public static boolean exists(String key) {
        Boolean result = null;
        ShardedJedis client = getInstance();
        try {
            result = client.exists(key);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return result;
    }

    /**
     * expire reset
     *
     * @param key
     * @param seconds 存活时间,单位/秒
     * @return Integer reply, specifically:
     * 1: the timeout was set.
     * 0: the timeout was not set since the key already has an associated timeout (versions lt 2.1.3), or the key does not exist.
     */
    public static long expire(String key, int seconds) {
        Long result = null;
        ShardedJedis client = getInstance();
        try {
            result = client.expire(key, seconds);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return result;
    }

    /**
     * expire at unixTime
     *
     * @param key
     * @param unixTime
     * @return
     */
    public static long expireAt(String key, long unixTime) {
        Long result = null;
        ShardedJedis client = getInstance();
        try {
            result = client.expireAt(key, unixTime);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return result;
    }

    public static boolean setnx(String key, String value, int seconds) {
        boolean boolResult = false;
        ShardedJedis client = getInstance();
        try {
            Long longResult = client.setnx(key, value);
            if (null != longResult && 1L == longResult) {
                client.expire(key, seconds);
                boolResult = true;
            }
        } catch (Exception e) {
            logger.error("setnx error", e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return boolResult;
    }

    public static void main(String[] args) {
        String xxlSsoRedisAddress = "redis://xxl-sso:password@127.0.0.1:6379/0";
        xxlSsoRedisAddress = "redis://127.0.0.1:6379/0";
        init(xxlSsoRedisAddress);

        setObjectValue("key", "666", 2 * 60 * 60);
        System.out.println(getObjectValue("key"));

    }

}
